package io.github.bigbird0101.datatransfer.container;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import integration.data.task.api.channel.Channel;
import integration.data.task.api.channel.impl.MemoryChannel;
import integration.data.task.api.exchanger.impl.RecordExchanger;
import integration.data.task.api.runner.AbstractRunner;
import integration.data.task.api.runner.ReaderRunner;
import integration.data.task.api.runner.WriterRunner;
import integration.data.task.api.transformation.impl.TransformerExecution;
import integration.data.task.api.utils.TransformerUtil;
import integration.data.task.reader.Reader;
import integration.data.task.writer.Writer;
import integration.modle.exception.IntegrationValidException;
import integration.utils.trace.LogUtils;
import io.github.bigbird0101.datatransfer.collector.Communication;
import io.github.bigbird0101.datatransfer.collector.StdoutPluginCollector;
import io.github.bigbird0101.datatransfer.collector.TaskPluginCollector;
import io.github.bigbird0101.datatransfer.constants.PluginType;
import io.github.bigbird0101.datatransfer.constants.State;
import io.github.bigbird0101.datatransfer.model.config.TransformerConfig;
import io.github.bigbird0101.datatransfer.record.RecordReceiver;
import lombok.extern.slf4j.Slf4j;

import java.util.List;

import static integration.data.task.api.constant.Constant.*;

/**
 * 任务
 */
@Slf4j
public class TaskExecutor {

    /*
    1、获取到页面配置
    -------
    2、创建一个任务容器
    3、配置转换
    4、读写端类型
    5、执行运行前置操作
    6、创建读写端
    7、数据转换
    8、运行
     */
    private String taskId;

    private String config;

    private Channel channel;

    /**
     * 输入端
     */
    private String readerName;

    private String readerConfig;


    /**
     * 转换
     */
    private String transformer;

    /**
     * 输出端
     */
    private String writerName;

    private String writerConfig;

    /**
     * 限速配置
     */
    private String channelConfig;

    private RecordExchanger recordSender;

    private Thread readerThread;

    private Thread writerThread;

    private ReaderRunner readerRunner;

    private WriterRunner writerRunner;

    /**
     * 该处的taskCommunication在多处用到：
     * 1. channel
     * 2. reader和writer的taskPluginCollector
     */
    private Communication taskCommunication;

    public TaskExecutor(String config) {

        if (StrUtil.isBlank(config)) {
            throw new IntegrationValidException("数据同步配置参数不能为空");
        }

        this.config = config;
        JSONObject configJson = JSONUtil.toBean(config, JSONObject.class);
        readerName = configJson.getByPath(READER_NAME, String.class);
        writerName = configJson.getByPath(WRITER_NAME, String.class);
        taskId = configJson.getStr(TASK_ID);

        Object reader = configJson.getByPath(READER_PARAMETER);
        Object writer = configJson.getByPath(WRITER_PARAMETER);

        Object transformation = configJson.get(TRANSFORMATION);

        Object channelSetting = configJson.getByPath(CHANNEL_SETTING);

        if (readerName == null) {
            throw new IntegrationValidException("来源端类型为空");
        }
        if (writerName == null) {
            throw new IntegrationValidException("目标端类型为空");
        }
        if (reader == null) {
            throw new IntegrationValidException("来源端配置为空");
        }
        if (writer == null) {
            throw new IntegrationValidException("目标端配置为空");
        }
        readerConfig = JSONUtil.toJsonStr(reader);
        writerConfig = JSONUtil.toJsonStr(writer);

        if (transformation != null) {
            this.transformer = JSONUtil.toJsonStr(transformation);
        } else {
            this.transformer = EMPTY_ARRAY_JSON;
        }

        if (channelSetting != null) {
            this.channelConfig = JSONUtil.toJsonStr(channelSetting);
        } else {
            this.channelConfig = EMPTY_OBJECT_JSON;
        }

        this.taskCommunication = new Communication();
        this.channel = new MemoryChannel(this.channelConfig);
        this.channel.setCommunication(this.taskCommunication);

        LogUtils.info(log, "task config reader plugin: {}, reader config: {}", readerName, readerConfig);
        LogUtils.info(log, "task config writer plugin: {}, writer config: {}", writerName, writerConfig);
        LogUtils.info(log, "task config transformer config: {}", this.transformer);
    }

    public Communication getTaskCommunication() {
        return taskCommunication;
    }

    // 初始化 reader / writer 配置
    public void init() {
        LogUtils.info(log, "task init start");

        // 转换函数
        List<TransformerExecution> transformerExecutionList = null;
        if (StrUtil.isNotBlank(transformer)) {
            List<TransformerConfig> transformerConfigList = JSONUtil.toList(transformer, TransformerConfig.class);
            transformerExecutionList = TransformerUtil.buildTransformerInfo(transformerConfigList);
        }

        /**
         * 生成writerThread
         */
        writerRunner = (WriterRunner) generateRunner(PluginType.WRITER, null);
        this.writerThread = new Thread(writerRunner, String.format("%s-writer", this.taskId));

        /**
         * 生成readerThread
         */
        readerRunner = (ReaderRunner) generateRunner(PluginType.READER, transformerExecutionList);
        this.readerThread = new Thread(readerRunner, String.format("%s-reader", this.taskId));

        LogUtils.info(log, "task init end， reader: {}, writer:{}, transformer function:{}", readerName, writerName, transformerExecutionList);
    }


    // 开始读和写
    public void start() {
        this.writerThread.start();
        // reader没有起来，writer不可能结束
        if (!this.writerThread.isAlive() || this.taskCommunication.getState() == State.FAILED) {
            throw new IntegrationValidException("任务writer运行失败", this.taskCommunication.getThrowable());
        }

        this.readerThread.start();

        // 这里reader可能很快结束
        if (!this.readerThread.isAlive() && this.taskCommunication.getState() == State.FAILED) {
            // 这里有可能出现Reader线上启动即挂情况 对于这类情况 需要立刻抛出异常
            throw new IntegrationValidException("任务reader运行失败", this.taskCommunication.getThrowable());
        }
    }

    // 检查任务是否结束
    public boolean isTaskFinished() {
        // 如果reader 或 writer没有完成工作，那么直接返回工作没有完成
        if (readerThread.isAlive() || writerThread.isAlive()) {
            return false;
        }

        if (taskCommunication == null || !taskCommunication.isFinished()) {
            return false;
        }
        return true;
    }

    public void stop() {
        recordSender.terminate();
        recordSender.shutdown();
    }

    /**
     * 强制停止任务
     */
    public void shutdown() {
        writerRunner.shutdown();
        readerRunner.shutdown();
        if (writerThread.isAlive()) {
            writerThread.interrupt();
        }
        if (readerThread.isAlive()) {
            readerThread.interrupt();
        }
    }

    private AbstractRunner generateRunner(PluginType pluginType, List<TransformerExecution> transformerInfoExecs) {
        AbstractRunner newRunner = null;
        TaskPluginCollector pluginCollector;
        switch (pluginType) {
            case READER:
                Reader reader = ReaderProcessor.getReaderName(readerName);
                // 创建读线程
                newRunner = new ReaderRunner(reader);
                // 读取端配置
                newRunner.setJobConf(readerConfig);
                // 脏数据收集器
                pluginCollector = new StdoutPluginCollector(config, taskCommunication, PluginType.READER);
                // 数据发送器
                recordSender = new RecordExchanger(taskId, channel, taskCommunication, transformerInfoExecs, pluginCollector);
                ((ReaderRunner) newRunner).setRecordSender(recordSender);
                /**
                 * 设置taskPlugin的collector，用来处理脏数据和job/task通信
                 */
                newRunner.setTaskPluginCollector(pluginCollector);

                break;
            case WRITER:
                Writer writer = WriterProcessor.getWriterName(writerName);
                // 创建写线程
                newRunner = new WriterRunner(writer);
                // 写入端配置
                newRunner.setJobConf(writerConfig);
                // 脏数据收集器
                pluginCollector = new StdoutPluginCollector(config, taskCommunication, PluginType.WRITER);
                RecordReceiver recordReceiver = new RecordExchanger(taskId, this.channel, pluginCollector);
                ((WriterRunner) newRunner).setRecordReceiver(recordReceiver);

                /**
                 * 设置taskPlugin的collector，用来处理脏数据和job/task通信
                 */
                newRunner.setTaskPluginCollector(pluginCollector);
                break;
            default:
                throw new IntegrationValidException("生成任务线程，不支持插件类型:" + pluginType);
        }
        newRunner.setTaskId(taskId);
        newRunner.setRunnerCommunication(this.taskCommunication);

        return newRunner;
    }


}
